package spark_core.operate_transform.keyvalue;

import java.util.Arrays;
import org.apache.spark.HashPartitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;
/**
 * @author shihb
 * @date 2020/1/8 11:43
 * 根据key,重新洗牌再分区,
 * 洗牌再分区的方式有hash,range也可以自定义
 */

public class PartitionByDemo {

  public static void main(String[] args) {
    //local模式,创建SparkConf对象设定spark的部署环境
    SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("mark rdd");
    //创建spark上下文对象（这边是java上下文）
    JavaSparkContext sc = new JavaSparkContext(sparkConf);



    //key-value的方法要pairRdd才能调用,所以创建JavaPairRDD
    JavaPairRDD inputPairRDD = sc.parallelizePairs(Arrays
            .asList(new Tuple2(1, "aa"), new Tuple2(2, "bb"), new Tuple2(3, "cc"), new Tuple2(4, "dd")),
        4);

    //参数为分区器,有hash分区器,range分区器,也可以自定义分区器
    //hash分区器,默认分区器,可能导致数据倾斜
    //range分区器,尽量保证数据量均匀,但要求数据可以排序和比较,所以spark使用较少
    JavaPairRDD<Integer, String> partitionByRdd = inputPairRDD
        .partitionBy(new HashPartitioner(2));

    partitionByRdd.glom().collect().forEach(System.out::println);


    //停止
    sc.stop();

  }
}
